#include <basic/errcode.h>
#include <basic/datetime.h>
#include <basic/ep.h>

#if defined(__linux__)
#include <sys/epoll.h>
#elif defined(__FreeBSD__)
#include <sys/event.h>
#endif

#define _READ_CACHE_DEF_SIZE    KB(1)   /* 读缓冲区大小 */
#define _READ_BUF_SIZE          KB(64)  /* 读临时区大小 */

struct _write_node
{/*{{{*/
    char *msg;
    int msg_size;
    int msg_written_size;
    struct listp_link link;
};/*}}}*/

struct _read_cache
{/*{{{*/
    char *msg;
    int msg_size;
    int msg_read_size;
};/*}}}*/

struct _ep_slot
{/*{{{*/
    int slot_idx;
    unsigned int is_listen;
    unsigned int is_wrap;
    SOCKET sock;
    struct sock_info sock_info;
    int timeout;
    time_t last_ts;
    struct _read_cache read_cache;
    struct listp_link write_list;
};/*}}}*/

struct _ep
{/*{{{*/
    struct _slot
    {
        unsigned int seq;
        struct _ep_slot *slot;
    } *slots;
    int slot_num;
    int slot_upper_idx;
    struct pollfd events[EP_POLL_MAX_NUM];
    int events_slot[EP_POLL_MAX_NUM];
    char read_buf[_READ_BUF_SIZE];
};/*}}}*/

static struct ep_active_slot *_create_active_slot(unsigned int type,
        struct _ep_slot *ep_slot, unsigned int seq)
{/*{{{*/
    struct ep_active_slot *node;

    ALLOC_RETERR(node, struct ep_active_slot, NULL);
    node->type=type;
    switch(type)
    {
        case EP_SLOT_CONN:
            break;
        case EP_SLOT_READ:
            node->_u._read.msg=NULL;
            node->_u._read.msg_size=0;
            break;
        case EP_SLOT_ABORT:
            break;
        case EP_SLOT_TIMEOUT:
            break;
    }
    node->slot=ep_slot;
    node->slot_idx=ep_slot->slot_idx;
    node->seq=seq;
    node->next=NULL;
    return(node);
}/*}}}*/

static void _append_active_list(struct ep_active_slot **head,
       struct ep_active_slot **tail, struct ep_active_slot *active)
{/*{{{*/
    if(active)
    {
        if(*head)
            (*tail)->next=active;
        else
            *head=active;

        while(active)
        {
            *tail=active;
            active=active->next;
        }
    }
}/*}}}*/

static void _slot_revoke(struct _ep *ep, struct _slot *slot)
{/*{{{*/
    struct _ep_slot *ep_slot=slot->slot;
    int revoke_idx=ep_slot->slot_idx;

    sock_close(ep_slot->sock);
    FREE(ep_slot->read_cache.msg);
    listp_free_memb(&ep_slot->write_list, struct _write_node, link,
            msg);
    FREE(slot->slot);
    if(revoke_idx == ep->slot_upper_idx-1)
    {
        int slot_idx;
        for(slot_idx=revoke_idx;
                slot_idx>=0 && !ep->slots[slot_idx].slot;
                slot_idx--)
        {   ep->slot_upper_idx--;   }
    }
}/*}}}*/

#define _ABORT(_ep,_slot,_error)                                \
do                                                              \
{                                                               \
    struct ep_active_slot *_active;                             \
    if((_active=_create_active_slot(EP_SLOT_ABORT,              \
                    _slot->slot, _slot->seq++)))                \
    {                                                           \
        _active->_u._abort.sock=_slot->slot->sock;              \
        _active->_u._abort.error=_error;                        \
        _append_active_list(&head, &tail, _active);             \
    }                                                           \
    _slot_revoke(_ep, _slot);                                   \
} while(0)

static struct ep_active_slot *_fill(struct _ep *ep,
        struct _slot *slot)
{/*{{{*/
    struct _ep_slot *ep_slot=slot->slot;
    char *read_buf;
    int read_num, num, read_offset;
    struct ep_active_slot *head, *tail, *active;
    int msg_size, msg_read_size, msg_need_size;

    head=tail=NULL;

    if(ep_slot->is_listen)
    {
        SOCKET newsock;
        if((newsock=sock_accept(ep_slot->sock, -1)) != INVALID_SOCKET)
        {
            if((active=_create_active_slot(EP_SLOT_CONN, slot->slot,
                            slot->seq++)))
            {
                active->_u._conn.sock=newsock;
                _append_active_list(&head, &tail, active);
            }
        }
        return(head);
    }

    // 读取消息，填充活跃消息列表
    read_buf=ep->read_buf;
    do
    {
        read_num=recv(ep_slot->sock, read_buf, _READ_BUF_SIZE, 0);
        if(read_num == SOCKET_ERROR)
        {
            int error=get_last_error(NULL);
            if(error == EINTR)  continue;
            if(error != EAGAIN)
            {
                //printf("RECV slot_idx(%d) abort, errno(%d)\n",
                //        slot->slot_idx, error);
                _ABORT(ep, slot, error);
            }
            break;
        }
        // 套接口接收FIN
        if(read_num == 0)
        {
            int error=get_last_error(NULL);
            //printf("%d FIN abort, errno(%d)\n",
            //slot->slot_idx, error);
            _ABORT(ep, slot, error);
            break;
        }

        // 有消息
        num=read_num;
        read_offset=0;
        if(ep_slot->is_wrap)
        {
            struct _read_cache *cache=&ep_slot->read_cache;
            while(num > 0)
            {
                if(!cache->msg)
                {
                    int tmp_size;
                    ALLOC_ARRAY(cache->msg, char, _READ_CACHE_DEF_SIZE,
                            tmp_size, FREE_CACHE);
                    cache->msg_size=-1;
                    cache->msg_read_size=0;
                }

                msg_size=cache->msg_size;
                msg_read_size=cache->msg_read_size;
                msg_need_size=((msg_size == -1) ? MSG_SIZE_SIZE :
                    msg_size) - msg_read_size;
                if(num < msg_need_size) // 现有内容不足以完成cache
                {
                    memcpy(cache->msg+msg_read_size,
                            read_buf+read_offset, num);
                    read_offset+=num;
                    cache->msg_read_size+=num;
                    //printf("1: num:%d msg_need_size:%d\n",
                    //        num, msg_need_size);
                    break;
                }

                //if(msg_size == -1)
                //{
                //    printf("2: num:%d read_size:%d need_size:%d\n", num, msg_read_size, msg_need_size);
                //}
                //else
                //{
                //    printf("3: num:%d read_size:%d need_size:%d\n", num, msg_read_size, msg_need_size);
                //    if(cache->msg_alloc_size == _READ_CACHE_DEF_SIZE &&
                //            msg_read_size+msg_need_size+1 >
                //            _READ_CACHE_DEF_SIZE)
                //    {
                //        EXPAND_ARRAY(cache->msg, char,
                //                msg_read_size+msg_need_size+1-
                //                _READ_CACHE_DEF_SIZE,
                //                cache->msg_alloc_size, FREE_CACHE);
                //        printf("4: num:%d read_size:%d need_size:%d\n", num, msg_read_size, msg_need_size);
                //    }
                //}
                memcpy(cache->msg+msg_read_size,
                        read_buf+read_offset, msg_need_size);
                read_offset+=msg_need_size;
                cache->msg_read_size+=msg_need_size;
                if(msg_size == -1)   // 上次未读完WRAP长度
                {
                    //printf("2: num:%d read_size:%d need_size:%d\n",
                    //        num, msg_read_size, msg_need_size);
                    cache->msg[MSG_SIZE_SIZE]=ZERO;
                    if(!str_is_hex(cache->msg) ||
                            (cache->msg_size=
                             strtol(cache->msg, NULL, 16)) >
                            SOCK_MSG_SIZE ||
                            cache->msg_size < 0)
                    {
                        _ABORT(ep, slot, -1);
                        goto FREE_CACHE;
                    }
                    cache->msg_read_size=0;
                    if(cache->msg_size+1 > _READ_CACHE_DEF_SIZE)
                    {
                        int tmp_size=_READ_CACHE_DEF_SIZE;
                        EXPAND_ARRAY(cache->msg, char,
                                cache->msg_size+1-_READ_CACHE_DEF_SIZE,
                                tmp_size, FREE_CACHE);
                        //printf("3: num:%d msg_size:%d\n",
                        //        num, cache->msg_size);
                    }
                }
                else    // 接收完整消息
                {
                    //printf("4: num:%d read_size:%d need_size:%d\n",
                    //        num, msg_read_size, msg_need_size);
                    if((active=_create_active_slot(EP_SLOT_READ, slot->slot,
                                    slot->seq++)))
                    {
                        active->_u._read.msg=cache->msg;
                        active->_u._read.msg_size=cache->msg_size;
                        //active->_u._read.check=slot->check;
                        cache->msg=NULL;
                        _append_active_list(&head, &tail, active);
                    }
                    else
                        goto FREE_CACHE;
                }
                num-=msg_need_size;
            }
        }
        else
        {
            struct ep_active_slot_read *nowrap_read;
            if(head)    // 合并至前读取节点 
            {
                nowrap_read=&head->_u._read;
                EXPAND_ARRAY(nowrap_read->msg, char, num,
                        nowrap_read->msg_size, FREE_HEAD);
                memcpy(nowrap_read->msg+nowrap_read->msg_size-num,
                        read_buf+read_offset, num);
            }
            else
            {
                if((head=_create_active_slot(EP_SLOT_READ, slot->slot,
                                slot->seq++)))
                {
                    nowrap_read=&head->_u._read;
                    ALLOC_ARRAY(nowrap_read->msg, char, num+1,
                            nowrap_read->msg_size, FREE_HEAD);
                    memcpy(nowrap_read->msg, read_buf, num);
                    nowrap_read->msg_size=num;
                }
                else
                    return(NULL);
            }
        }
    } while(read_num == _READ_BUF_SIZE);

    return(head);

FREE_CACHE:
    FREE(ep_slot->read_cache.msg);
    return(head);

FREE_HEAD:
    FREE(head->_u._read.msg);
    FREE(head);
    return(NULL);
}/*}}}*/

static int _flush(struct _slot *slot)
{/*{{{*/
    struct _ep_slot *ep_slot;
    struct _write_node *write_node;
    int num;
    int ret=0;

    if(!(ep_slot=slot->slot))
        return(-1);

    while(!listp_is_empty(&ep_slot->write_list))
    {
        write_node=listp_entry(ep_slot->write_list.next,
             struct _write_node, link);
AGAIN:
        num=send(ep_slot->sock,
                write_node->msg+write_node->msg_written_size,
                write_node->msg_size-write_node->msg_written_size,
                0);
		if(num == SOCKET_ERROR)
		{
            int errnum=get_last_error(NULL);
            if(errnum == EINTR)
                goto AGAIN;
            if(errnum != EAGAIN)
                ret=-1;
            goto OUT_GO;
		}
        write_node->msg_written_size+=num;
        if(write_node->msg_size > write_node->msg_written_size)
            goto OUT_GO;
        listp_del(ep_slot->write_list.next);
        FREE(write_node->msg);
        FREE(write_node);
    }

OUT_GO:
    return(ret); 
}/*}}}*/

int ep_init()
{/*{{{*/
    return(0);
}/*}}}*/

int ep_clean()
{/*{{{*/
    return(0);
}/*}}}*/

int ep_create(EP *epp)
{/*{{{*/
    int slot_idx;
    struct _ep *ep;
    struct _slot *slot;

    if(!epp)
        return(RET_ERR_PARA);
    *epp=(struct _ep *)NULL;

    ALLOC_RETERR(ep, struct _ep, RET_ERR_ALLOC);
    ep->slot_num=EP_POLL_MAX_NUM;
    ALLOC_MULTI_DOERR(ep->slots, struct _slot, ep->slot_num,
            free(ep); return(RET_ERR_ALLOC));
    for(slot_idx=0; slot_idx<ep->slot_num; slot_idx++)
    {
        slot=&ep->slots[slot_idx];
        //slot->check=1;
        slot->seq=0;
        slot->slot=(struct _ep_slot *)NULL;
    }
    ep->slot_upper_idx=0;

    *epp=ep;
    return(0);
}/*}}}*/

int ep_free(EP ep)
{/*{{{*/
    int i;

    if(!ep)
        return(RET_ERR_PARA);
    for(i=0; i<ep->slot_upper_idx; i++)
        ep_slot_revoke(ep, ep->slots[i].slot);
    free(ep->slots);
    free(ep);
    return(0);
}/*}}}*/

int ep_slot_set_wrap(EP_SLOT ep_slot)
{/*{{{*/
    if(ep_slot)
    {
        if(ep_slot->is_listen)
        {   return(-1);   }
        else
        {
            ep_slot->is_wrap=IS_TRUE;
            return(0);
        }
    }
    else
        return(RET_ERR_PARA);
}/*}}}*/

int ep_slot_set_timeout(EP_SLOT ep_slot, int timeout)
{/*{{{**/
    if(ep_slot)
    {
        if(ep_slot->is_listen)
        {   return(-1);   }
        else
        {
            ep_slot->timeout=timeout;
            ep_slot->last_ts=curr_timestamp(NULL);
            return(0);
        }
    }
    else
        return(RET_ERR_PARA);
}/*}}}*/

int ep_slot_register(EP ep, SOCKET sock, unsigned int is_listen,
        EP_SLOT *p_ep_slot)
{/*{{{*/
	int slot_idx;
    struct _slot *slot=(struct _slot *)NULL;
    struct _ep_slot *ep_slot=(struct _ep_slot *)NULL;

    if(!ep || sock < 0)
        return(RET_ERR_PARA);

	for(slot_idx=0; slot_idx<ep->slot_upper_idx; slot_idx++)
    {
        slot=&ep->slots[slot_idx];
		if(!slot->slot)
            break;
    }
	if(slot_idx == ep->slot_upper_idx)
    {
        if(ep->slot_upper_idx == ep->slot_num)
            return(-1);
        slot=&ep->slots[slot_idx];
        ep->slot_upper_idx++;
    }

    ALLOC_RETERR(slot->slot, struct _ep_slot, RET_ERR_ALLOC);
    ep_slot=slot->slot;
    ep_slot->slot_idx=slot_idx;
	ep_slot->sock=sock;
    sock_get_info(sock, &ep_slot->sock_info);
    if(is_listen)
    {
        ep_slot->is_listen=IS_TRUE;
        CLR_BUF(ep_slot->sock_info.peer_ip);
        ep_slot->sock_info.peer_port=-1;
    }
    else
    {
        ep_slot->is_listen=IS_FALSE;
        sock_set_nonblock(ep_slot->sock);
    }
    ep_slot->is_wrap=IS_FALSE;
    ep_slot->timeout=0;
    ep_slot->last_ts=curr_timestamp(NULL);
    ep_slot->read_cache.msg=(char *)NULL;
    listp_init(&ep_slot->write_list);
    //if(++slot->check == 0)
    //    slot->check=1;

    if(p_ep_slot) *p_ep_slot=ep_slot;
    return(0);
}/*}}}*/

int ep_slot_revoke(EP ep, EP_SLOT ep_slot)
{/*{{{*/
    struct _slot *slot;

    if(!ep || !ep_slot ||
            ep_slot->slot_idx < 0 ||
            ep_slot->slot_idx > ep->slot_num-1)
        return(RET_ERR_PARA); 

    slot=&ep->slots[ep_slot->slot_idx];
    if(ep_slot == slot->slot)
    {
        _slot_revoke(ep, slot);
        return(0);
    }
    else
        return(-1);
}/*}}}*/

int ep_poll(EP ep, struct ep_active_slot **active_list)
{/*{{{*/
#define _POLL_TIMEOUT_MS     500
    int poll_num;
    int event_num;
    int event_idx;
    int slot_idx;
    struct ep_active_slot *head, *tail;
    struct _slot *slot;
    struct _ep_slot *ep_slot;
    struct pollfd *fd;
    time_t now;

    if(!ep || !active_list)
        return(RET_ERR_PARA); 

    *active_list=head=tail=NULL;

    now=curr_timestamp(NULL);
    for(slot_idx=0, event_num=0; slot_idx<ep->slot_upper_idx;
            slot_idx++)
    {
        slot=&ep->slots[slot_idx];
        if(!(ep_slot=slot->slot)) continue;
        if(!ep_slot->is_listen && ep_slot->timeout > 0 &&
                    now - ep_slot->last_ts > ep_slot->timeout)
        {
            struct ep_active_slot *active;
            if((active=_create_active_slot(EP_SLOT_TIMEOUT, slot->slot,
                            slot->seq++)))
            {
                active->_u._timeout.sock=ep_slot->sock;
                _append_active_list(&head, &tail, active);
            }
            _slot_revoke(ep, slot);
            continue;
        }

        fd=&ep->events[event_num];
        fd->events=0;
        fd->revents=0;
        fd->events |= POLLIN;
        if(!listp_is_empty(&ep_slot->write_list))
            fd->events |= POLLOUT;
        fd->fd=ep_slot->sock;
        ep->events_slot[event_num]=slot_idx;
        event_num++;
    }

POLL_AGAIN:
    poll_num=poll(ep->events, event_num, _POLL_TIMEOUT_MS);
    if(poll_num == SOCKET_ERROR)
    {
        if(get_last_error(NULL) == EINTR)
            goto POLL_AGAIN;
        return(-1);
    }
    if(poll_num == 0) goto OUT_GO;

    now++;
    for(event_idx=0; event_idx<event_num; event_idx++)
    {
        fd=&ep->events[event_idx];
        slot_idx=ep->events_slot[event_idx];
        slot=&ep->slots[slot_idx];
        if(slot->slot->timeout > 0)
            slot->slot->last_ts=now;
        if(fd->revents)
        {
#if defined(__APPLE__)
            if(fd->revents & POLLHUP)
            {
                _ABORT(ep, slot, 0);
                continue;
            }
#endif
            if(fd->revents & POLLIN)
                _append_active_list(&head, &tail,
                        _fill(ep, slot));
            if(fd->revents & POLLOUT)
                _flush(slot);
        }
    }

OUT_GO:
    *active_list=head;
    return(0);
}/*}}}*/
        
int ep_write(EP ep, EP_SLOT ep_slot, char *msg, int *msg_size)
{/*{{{*/
    struct _slot *slot;
    int msg_total_size, msg_left_size;
	int written_num;
    struct _write_node *write_node;

    if(!ep || !ep_slot || !msg || !msg_size)
        return(RET_ERR_PARA);
    if(ep_slot->slot_idx < 0 ||
            ep_slot->slot_idx > ep->slot_num-1)
        return(RET_ERR_PARA);
    slot=&ep->slots[ep_slot->slot_idx];
    if(ep_slot != slot->slot)
        return(-1);
    msg_total_size=msg_left_size=*msg_size;
    if(msg_total_size == 0)  return(0);
    if(msg_total_size < 0 || msg_total_size > SOCK_MSG_SIZE)
        return(-2); 

    if(ep_slot->is_listen) return(-3);
    if(ep_slot->is_wrap)
    {
        wrap_size((unsigned char *)msg, &msg_total_size,
                (unsigned char *)msg);
        msg_left_size=msg_total_size;
    }
    if(ep_slot->timeout > 0)
        ep_slot->last_ts=curr_timestamp(NULL);

	// 写队列为空
    written_num=0;
	if(listp_is_empty(&ep_slot->write_list))
	{
AGAIN:
        written_num=send(ep_slot->sock, msg, msg_left_size, 0);
		if(written_num == SOCKET_ERROR)
		{
            int error=get_last_error(NULL);
		    if(error == EINTR)
                goto AGAIN;
			if(error != EAGAIN)
            {
                //printf("WRITE error(%d) slot_idx(%d) msg_size(%d)\n", error, slot->slot_idx, *msg_size);
                _slot_revoke(ep, slot);
                return(-4);
            }
			written_num=0;
		}
		msg_left_size-=written_num;
		if(msg_left_size == 0) return(0);
	}

	// 加入写队列
    ALLOC_RETERR(write_node, struct _write_node, RET_ERR_ALLOC);
    ALLOC_MULTI_DOERR(write_node->msg, char, msg_left_size+1,
           free(write_node); return(RET_ERR_ALLOC));
    memcpy(write_node->msg, msg+written_num, msg_left_size);
    write_node->msg_size=msg_left_size;
    write_node->msg_written_size=0;
    listp_add_tail(&write_node->link, &ep_slot->write_list);

    return(0);
}/*}}}*/

int ep_get_info(EP ep, struct ep_info *info)
{/*{{{*/
    struct _slot *slot;
    int used_slot_num;
    int slot_idx;

    if(!ep || !info)
        return(RET_ERR_PARA);

    info->slot_num=ep->slot_num;
    used_slot_num=0;
    for(slot_idx=0; slot_idx<ep->slot_upper_idx; slot_idx++)
    {
        slot=&ep->slots[slot_idx];
        if(slot->slot) used_slot_num++;
    }
    info->used_slot_num=used_slot_num;
    return(0);
}/*}}}*/

int ep_get_slot_info(EP ep, EP_SLOT ep_slot,
        struct ep_slot_info *slot_info)
{/*{{{*/
    struct _slot *slot;

    if(!ep || !ep_slot || !slot_info)
        return(RET_ERR_PARA);
    if(ep_slot->slot_idx < 0 ||
            ep_slot->slot_idx > ep->slot_num-1)
        return(RET_ERR_PARA);
    slot=&ep->slots[ep_slot->slot_idx];
    if(ep_slot == slot->slot)
    {
        slot_info->slot_idx=ep_slot->slot_idx;
        slot_info->sock=ep_slot->sock;
        slot_info->is_listen=ep_slot->is_listen;
        slot_info->is_wrap=ep_slot->is_wrap;
        memcpy(&slot_info->sock_info, &ep_slot->sock_info,
                sizeof(struct sock_info));
        slot_info->seq=slot->seq;
    }
    else
    {
        slot_info->slot_idx=-1;
        slot_info->sock=INVALID_SOCKET;
        slot_info->is_listen=IS_FALSE;
        slot_info->is_wrap=IS_FALSE;
        CLR_BUF(slot_info->sock_info.self_ip);
        slot_info->sock_info.self_port=0;
        CLR_BUF(slot_info->sock_info.peer_ip);
        slot_info->sock_info.peer_port=0;
        slot_info->seq=0;
    }
    return(0);
}/*}}}*/

#if 0
//=================================================//

#define _THRD_MAX_NUM           32      /* 并行线程最大数量 */
#define _FETCH_EVENT_NUM        8       /* 一次获取通知数量 */

struct _epx_thrd
{/*{{{*/
#if defined(__linux__)
    struct epoll_event events[_FETCH_EVENT_NUM];
#elif defined(__FreeBSD__) || defined(__APPLE__)
    struct kevent events[_FETCH_EVENT_NUM];
#endif
    int event_num;
    char read_buf[_READ_BUF_SIZE];
};/*}}}*/

#if 0
struct _epx_slot
{/*{{{*/
    SOCKET sock;
    struct sock_info sock_info;
    unsigned int is_listen;
    unsigned int is_wrap;
    int slot_idx;
    int timeout;
    time_t last_ts;
    struct _read_cache read_cache;
    struct listp_link write_list;
};/*}}}*/
#endif
#define _epx_slot _ep_slot

struct _epx
{/*{{{*/
    unsigned int is_multi_thread;
    int queue;
    struct _x_slot
    {
        pthread_mutex_t lock;
        unsigned short check;
        unsigned int seq;
        struct _epx_slot *slot;
    } *slots;
    int slot_num;
    struct _epx_thrd *single_thrd;
};/*}}}*/

#if 0
static int _thrd_register(struct _epx *epx, int thrd_idx,
        pthread_t thrd_id)
{/*{{{*/
    struct _thrd *thrd;

    ALLOC_RETERR(thrd, struct _thrd, -1); 
    thrd->thrd_id=thrd_id;
    thrd->thrd_idx=thrd_idx;
    ep->thrds[thrd_idx]=thrd;
    return(0);
}/*}}}*/


static void _thrd_revoke(struct _ep *epx, int thrd_idx)
{/*{{{*/
    FREE(epx->thrds[thrd_idx]);
}/*}}}*/

static int _get_thrd(struct _epx *epx)
{/*{{{*/
    if(epx->is_multi_thread)
    {
        pthread_t thrd_id=pthread_self();
        struct _thrd *thrd;
        int thrd_idx;

        for(thrd_idx=0; thrd_idx<_THRD_MAX_NUM; thrd_idx++)
        {
            if((thrd=ep->thrds[thrd_idx]) && thrd->thrd_id == thrd_id)
                return(thrd_idx);
        }
        return(-1);
    }
    else
        return(0);
}/*}}}*/
#endif

static void _x_slot_revoke(struct _epx *epx, struct _x_slot *slot)
{/*{{{*/
    struct _epx_slot *epx_slot=slot->slot;

#if defined(__linux__)
    struct epoll_event ev;

    ev.data.fd=epx_slot->slot_idx;
    ev.events=0;
    epoll_ctl(epx->queue, EPOLL_CTL_DEL, epx_slot->sock, &ev);
#elif defined(__FreeBSD__) || defined(__APPLE__)
    struct kevent changes[2];
    int nchanges;

    if(epx_slot->is_listen)
    {
        EV_SET(&changes[0], epx_slot->sock, EVFILT_READ, EV_DELETE,
                0, 0, 0);
        nchanges=1;
    }
    else
    {
        EV_SET(&changes[0], epx_slot->sock, EVFILT_READ, EV_DELETE,
                0, 0, 0);
        EV_SET(&changes[1], epx_slot->sock, EVFILT_WRITE, EV_DELETE,
                0, 0, 0);
        nchanges=2;
    }
    kevent(epx->queue, changes, nchanges, NULL, 0, NULL);
#endif

    sock_close(epx_slot->sock);
    FREE(epx_slot->read_cache.msg);
    listp_free_memb(&epx_slot->write_list, struct _write_node, link,
            msg);
    FREE(slot->slot);
}/*}}}*/

#define _X_ABORT(_epx,_slot,_error)                                 \
do                                                                  \
{                                                                   \
    struct ep_active_slot *_active;                                 \
    if((_active=_create_active_slot(EP_SLOT_ABORT, _slot->slot,     \
                    _slot->seq++)))                                 \
    {                                                               \
        _active->_u._abort.sock=_slot->slot->sock;                  \
        _active->_u._abort.error=_error;                            \
        _append_active_list(&head, &tail, _active);                 \
    }                                                               \
    _x_slot_revoke(_epx, _slot);                                    \
} while(0)

static struct ep_active_slot *_x_fill(struct _epx *epx,
        struct _epx_thrd *epx_thrd,
        struct _x_slot *slot)
{/*{{{*/
    struct _epx_slot *epx_slot=slot->slot;
    char *read_buf;
    int read_num, num, read_offset;
    struct ep_active_slot *head, *tail, *active;
    int msg_size, msg_read_size, msg_need_size;

    head=tail=NULL;

    if(epx_slot->is_listen)
    {
        SOCKET newsock;
        if((newsock=sock_accept(epx_slot->sock, -1)) != INVALID_SOCKET)
        {
            if((active=_create_active_slot(EP_SLOT_CONN, epx_slot,
                            slot->seq++)))
            {
                active->_u._conn.sock=newsock;
                _append_active_list(&head, &tail, active);
            }
        }
        return(head);
    }

    // 读取消息，填充活跃消息列表
    read_buf=epx_thrd->read_buf;
    do
    {
        read_num=recv(epx_slot->sock, read_buf, _READ_BUF_SIZE, 0);
        if(read_num == SOCKET_ERROR)
        {
            int error=get_last_error(NULL);
            if(error == EINTR)  continue;
            if(error != EAGAIN)
            {
                //printf("RECV slot_idx(%d) abort, errno(%d)\n",
                //        slot->slot_idx, error);
                _X_ABORT(epx, slot, error);
            }
            break;
        }
        // 套接口接收FIN
        if(read_num == 0)
        {
            int error=get_last_error(NULL);
            //printf("%d FIN abort, errno(%d)\n",
            //slot->slot_idx, error);
            _X_ABORT(epx, slot, error);
            break;
        }

        // 有消息
        num=read_num;
        read_offset=0;
        if(epx_slot->is_wrap)
        {
            struct _read_cache *cache=&epx_slot->read_cache;
            while(num > 0)
            {
                if(!cache->msg)
                {
                    int tmp_size;
                    ALLOC_ARRAY(cache->msg, char, _READ_CACHE_DEF_SIZE,
                            tmp_size, FREE_CACHE);
                    cache->msg_size=-1;
                    cache->msg_read_size=0;
                }

                msg_size=cache->msg_size;
                msg_read_size=cache->msg_read_size;
                msg_need_size=((msg_size == -1) ? MSG_SIZE_SIZE :
                    msg_size) - msg_read_size;
                if(num < msg_need_size) // 现有内容不足以完成cache
                {
                    memcpy(cache->msg+msg_read_size,
                            read_buf+read_offset, num);
                    read_offset+=num;
                    cache->msg_read_size+=num;
                    //printf("1: num:%d msg_need_size:%d\n",
                    //        num, msg_need_size);
                    break;
                }

                //if(msg_size == -1)
                //{
                //    printf("2: num:%d read_size:%d need_size:%d\n", num, msg_read_size, msg_need_size);
                //}
                //else
                //{
                //    printf("3: num:%d read_size:%d need_size:%d\n", num, msg_read_size, msg_need_size);
                //    if(cache->msg_alloc_size == _READ_CACHE_DEF_SIZE &&
                //            msg_read_size+msg_need_size+1 >
                //            _READ_CACHE_DEF_SIZE)
                //    {
                //        EXPAND_ARRAY(cache->msg, char,
                //                msg_read_size+msg_need_size+1-
                //                _READ_CACHE_DEF_SIZE,
                //                cache->msg_alloc_size, FREE_CACHE);
                //        printf("4: num:%d read_size:%d need_size:%d\n", num, msg_read_size, msg_need_size);
                //    }
                //}
                memcpy(cache->msg+msg_read_size,
                        read_buf+read_offset, msg_need_size);
                read_offset+=msg_need_size;
                cache->msg_read_size+=msg_need_size;
                if(msg_size == -1)   // 上次未读完WRAP长度
                {
                    //printf("2: num:%d read_size:%d need_size:%d\n",
                    //        num, msg_read_size, msg_need_size);
                    cache->msg[MSG_SIZE_SIZE]=ZERO;
                    if(!str_is_hex(cache->msg) ||
                            (cache->msg_size=
                             strtol(cache->msg, NULL, 16)) >
                            SOCK_MSG_SIZE ||
                            cache->msg_size < 0)
                    {
                        _X_ABORT(epx, slot, -1);
                        goto FREE_CACHE;
                    }
                    cache->msg_read_size=0;
                    if(cache->msg_size+1 > _READ_CACHE_DEF_SIZE)
                    {
                        int tmp_size=_READ_CACHE_DEF_SIZE;
                        EXPAND_ARRAY(cache->msg, char,
                                cache->msg_size+1-_READ_CACHE_DEF_SIZE,
                                tmp_size, FREE_CACHE);
                        //printf("3: num:%d msg_size:%d\n",
                        //        num, cache->msg_size);
                    }
                }
                else    // 接收完整消息
                {
                    //printf("4: num:%d read_size:%d need_size:%d\n",
                    //        num, msg_read_size, msg_need_size);
                    if((active=_create_active_slot(EP_SLOT_READ,
                                    epx_slot, slot->seq++)))
                    {
                        active->_u._read.msg=cache->msg;
                        active->_u._read.msg_size=cache->msg_size;
                        //active->_u._read.check=slot->check;
                        cache->msg=NULL;
                        _append_active_list(&head, &tail, active);
                    }
                    else
                        goto FREE_CACHE;
                }
                num-=msg_need_size;
            }
        }
        else
        {
            struct ep_active_slot_read *nowrap_read;
            if(head)    // 合并至前读取节点 
            {
                nowrap_read=&head->_u._read;
                EXPAND_ARRAY(nowrap_read->msg, char, num,
                        nowrap_read->msg_size, FREE_HEAD);
                memcpy(nowrap_read->msg+nowrap_read->msg_size-num,
                        read_buf+read_offset, num);
            }
            else
            {
                if((head=_create_active_slot(EP_SLOT_READ, epx_slot,
                                slot->seq++)))
                {
                    nowrap_read=&head->_u._read;
                    ALLOC_ARRAY(nowrap_read->msg, char, num+1,
                            nowrap_read->msg_size, FREE_HEAD);
                    memcpy(nowrap_read->msg, read_buf, num);
                    nowrap_read->msg_size=num;
                }
                else
                    return(NULL);
            }
        }
    } while(read_num == _READ_BUF_SIZE);

    return(head);

FREE_CACHE:
    FREE(epx_slot->read_cache.msg);
    return(head);

FREE_HEAD:
    FREE(head->_u._read.msg);
    FREE(head);
    return(NULL);
}/*}}}*/

static int _x_flush(struct _x_slot *slot)
{/*{{{*/
    struct _epx_slot *epx_slot;
    struct _write_node *write_node;
    int num;

    if(!(epx_slot=slot->slot))
        return(-1);

    while(!listp_is_empty(&epx_slot->write_list))
    {
        write_node=listp_entry(epx_slot->write_list.next,
             struct _write_node, link);
AGAIN:
        num=send(epx_slot->sock,
                write_node->msg+write_node->msg_written_size,
                write_node->msg_size-write_node->msg_written_size,
                0);
		if(num == SOCKET_ERROR)
		{
            int errnum=get_last_error(NULL);
            if(errnum == EINTR)
                goto AGAIN;
            if(errnum != EAGAIN)
                return(-1);
		}
        write_node->msg_written_size+=num;
        if(write_node->msg_size > write_node->msg_written_size)
            return(0);
        listp_del(epx_slot->write_list.next);
        FREE(write_node->msg);
        FREE(write_node);
    }

    return(0); 
}/*}}}*/

int epx_init()
{/*{{{*/
	int max_nofile;
    struct rlimit lim;

    // 修改文件描述字限额
    getrlimit(RLIMIT_NOFILE, &lim);
    max_nofile=lim.rlim_cur;

    if(geteuid())
    {
        if(lim.rlim_cur < lim.rlim_max)
        {
            lim.rlim_cur=(unsigned int)lim.rlim_max & 0x7FFFFFFF;
            if(lim.rlim_cur > 10240)
                lim.rlim_cur=10240;
            if(setrlimit(RLIMIT_NOFILE, &lim) == 0)
                max_nofile=lim.rlim_cur;
        }
    }
    else
    {
        lim.rlim_cur=lim.rlim_max=10240;
        while(lim.rlim_cur > max_nofile)
        {
            if(setrlimit(RLIMIT_NOFILE, &lim) == 0)
            {
                max_nofile=lim.rlim_cur;
                break;
            }
            lim.rlim_cur/=2;
            lim.rlim_max/=2;
        }
    }

    return(0);
}/*}}}*/

int epx_clean()
{/*{{{*/
    return(0);
}/*}}}*/

int epx_create(unsigned int is_multi_thread, EPX *epxp)
{/*{{{*/
    int slot_idx;
    struct _epx *epx;
    struct _x_slot *slot;
    struct rlimit lim;

    if(!epxp)
        return(RET_ERR_PARA);
    *epxp=(struct _epx *)NULL;

    ALLOC_RETERR(epx, struct _epx, RET_ERR_ALLOC);
    getrlimit(RLIMIT_NOFILE, &lim);
    epx->slot_num=lim.rlim_cur & 0xfffffff0;
#if defined(__linux__)
    epx->queue=epoll_create(epx->slot_num);
#elif defined(__FreeBSD__) || defined(__APPLE__)
    epx->queue=kqueue();
#else
    epx->queue=-1;
#endif
    if(epx->queue == -1)
    {
        free(epx);
        return(-1);
    }
    ALLOC_MULTI_DOERR(epx->slots, struct _x_slot, epx->slot_num,
            free(epx); return(RET_ERR_ALLOC));
    for(slot_idx=0; slot_idx<epx->slot_num; slot_idx++)
    {
        slot=&epx->slots[slot_idx];
        if(is_multi_thread)
            pthread_mutex_init(&slot->lock, NULL);
        //slot->check=1;
        slot->seq=0;
        slot->slot=(struct _epx_slot *)NULL;
    }
    if(is_multi_thread)
    {
        epx->is_multi_thread=IS_TRUE;
        epx->single_thrd=(struct _epx_thrd *)NULL;
    }
    else
    {
        epx->is_multi_thread=IS_FALSE;
        ALLOC_DOERR(epx->single_thrd, struct _epx_thrd,
                free(epx->slots); free(epx); return(RET_ERR_ALLOC));
    }

    *epxp=epx;
    return(0);
}/*}}}*/

int epx_free(EPX epx)
{/*{{{*/
    int i;

    if(!epx)
        return(RET_ERR_PARA);

    for(i=0; i<epx->slot_num; i++)
        _x_slot_revoke(epx, &epx->slots[i]);
    free(epx->slots);
    if(epx->is_multi_thread)
    {}
    else
    {   free(epx->single_thrd);   }
    close(epx->queue);
    free(epx);

    return(0);
}/*}}}*/

int epx_thrd_create(EPX epx, EPX_THRD *p_epx_thrd)
{/*{{{*/
    if(!epx || !p_epx_thrd)
        return(RET_ERR_PARA);

    if(epx->is_multi_thread)
    {
        struct _epx_thrd *epx_thrd;
        ALLOC_RETERR(epx_thrd, struct _epx_thrd, RET_ERR_ALLOC); 
        *p_epx_thrd=epx_thrd;
        return(0);
    }
    else
        return(-1);
}/*}}}*/

int epx_slot_create(SOCKET sock, unsigned int is_listen,
        EPX_SLOT *p_epx_slot)
{/*{{{*/
    struct _epx_slot *epx_slot=NULL;

    if(!p_epx_slot)
        return(RET_ERR_PARA);
    *p_epx_slot=(struct _epx_slot *)NULL;
    if(sock < 0)
        return(RET_ERR_PARA);

    if(sock_set_nonblock(sock) == -1)
        return(-1);
    ALLOC_RETERR(epx_slot, struct _epx_slot, RET_ERR_ALLOC);
	epx_slot->sock=sock;
    sock_get_info(sock, &epx_slot->sock_info);
    if(is_listen)
    {
        epx_slot->is_listen=IS_TRUE;
        CLR_BUF(epx_slot->sock_info.peer_ip);
        epx_slot->sock_info.peer_port=-1;
    }
    else
    {
        epx_slot->is_listen=IS_FALSE;
    }
    epx_slot->is_wrap=IS_FALSE;
    epx_slot->slot_idx=-1;
    epx_slot->timeout=0;
    epx_slot->last_ts=0;
    epx_slot->read_cache.msg=(char *)NULL;
    listp_init(&epx_slot->write_list);

    *p_epx_slot=epx_slot;
    return(0);
}/*}}}*/

int epx_slot_set_wrap(EPX_SLOT epx_slot)
{/*{{{*/
    if(epx_slot)
    {
        if(epx_slot->is_listen)
        {   return(-1);   }
        else
        {
            epx_slot->is_wrap=IS_TRUE;
            return(0);
        }
    }
    else
        return(RET_ERR_PARA);
}/*}}}*/

int epx_slot_set_timeout(EPX_SLOT epx_slot, int timeout)
{/*{{{**/
    if(epx_slot)
    {
        if(epx_slot->is_listen)
        {   return(-1);   }
        else
        {
            epx_slot->timeout=timeout;
            epx_slot->last_ts=curr_timestamp(NULL);
            return(0);
        }
    }
    else
        return(RET_ERR_PARA);
}/*}}}*/

int epx_slot_register(EPX epx, EPX_SLOT epx_slot)
{/*{{{*/
#if defined(__linux__)
    struct epoll_event ev;
#elif defined(__FreeBSD__) || defined(__APPLE__)
    struct kevent changes[2];
    int nchanges;
#endif
    int ret;
    int slot_idx;
    struct _x_slot *slot=(struct _x_slot *)NULL;

    if(!epx || !epx_slot)
        return(RET_ERR_PARA);

    for(slot_idx=0; slot_idx<epx->slot_num; slot_idx++)
    {
        slot=&epx->slots[slot_idx];
        if(epx->is_multi_thread)
            pthread_mutex_lock(&slot->lock);
        if(!slot->slot)
            break;
        if(epx->is_multi_thread)
            pthread_mutex_unlock(&slot->lock);
    }
    if(slot_idx == epx->slot_num)
        return(-1);

#if defined(__linux__)
    if(epx_slot->is_listen)
        ev.events=EPOLLIN;
    else
        ev.events=EPOLLIN | EPOLLOUT | EPOLLET;
    ev.data.fd=slot_idx;
    if(epoll_ctl(epx->queue, EPOLL_CTL_ADD, epx_slot->sock, &ev) ==
            -1)
    {   ret=-2;   }
    else
#elif defined(__FreeBSD__) || defined(__APPLE__)
    if(epx_slot->is_listen)
    {
        EV_SET(&changes[0], epx_slot->sock, EVFILT_READ, EV_ADD, 0, 0,
                (void *)slot_idx);
        nchanges=1;
    }
    else
    {
        EV_SET(&changes[0], epx_slot->sock, EVFILT_READ, EV_ADD, 0, 0,
                (void *)slot_idx);
        EV_SET(&changes[1], epx_slot->sock, EVFILT_WRITE, EV_ADD, 0, 0,
                (void *)slot_idx);
        nchanges=2;
    }
    if(kevent(epx->u_poll.kq, changes, nchanges, NULL, 0,
                NULL) == -1)
    {   ret=-3;   }
    else
#endif
    {
        epx_slot->slot_idx=slot_idx;
        slot->slot=epx_slot;
        //if(++slot->check == 0)
        //    slot->check=1;
        ret=0;
    }
    if(epx->is_multi_thread)
        pthread_mutex_unlock(&slot->lock);

    return(ret);
}/*}}}*/

int epx_slot_revoke(EPX epx, EPX_SLOT epx_slot)
{/*{{{*/
    int ret;
    int slot_idx;
    struct _x_slot *slot;

    if(!epx || !epx_slot)
        return(RET_ERR_PARA); 
    if((slot_idx=epx_slot->slot_idx) < 0 || slot_idx > epx->slot_num)
        return(RET_ERR_PARA);

    slot=&epx->slots[slot_idx];
    if(epx->is_multi_thread) pthread_mutex_lock(&slot->lock);
    if(slot->slot == epx_slot)
    {
        _x_slot_revoke(epx, slot);
        ret=0;
    }
    else
        ret=-1;
    if(epx->is_multi_thread) pthread_mutex_unlock(&slot->lock);

    return(ret);
}/*}}}*/

int epx_poll(EPX epx, EPX_THRD epx_thrd,
        struct ep_active_slot **active_list)
{/*{{{*/
#define _POLL_TIMEOUT_MS     500
#if defined(__linux__)
    struct epoll_event *fd;
#elif defined(__FreeBSD__)
    struct timespec timeout;
#endif
    int poll_num;
    int event_idx;
    int slot_idx;
    struct _x_slot *slot;
    struct ep_active_slot *head, *tail;

    if(!epx || !active_list)
        return(RET_ERR_PARA); 
    if(epx->is_multi_thread)
    {
        if(!epx_thrd)
            return(RET_ERR_PARA); 
    }
    else
    {
        if(epx_thrd)
            return(RET_ERR_PARA); 
        epx_thrd=epx->single_thrd;
    }

    *active_list=head=tail=NULL;

#if defined(__linux__)
EPOLL_AGAIN:
    poll_num=epoll_wait(epx->queue, epx_thrd->events,
            _FETCH_EVENT_NUM, _POLL_TIMEOUT_MS);
    if(poll_num == SOCKET_ERROR)
    {
        if(get_last_error(NULL) == EINTR)
            goto EPOLL_AGAIN;
        return(-1);
    }
    if(poll_num == 0)  return(0);

    /*
       printf("thrd %u\n", (unsigned int)pthread_self());
       printf("poll_num: %d\n", poll_num); 
       printf("active_num: %d\n", active_num);
       */
    for(event_idx=0; event_idx<poll_num; event_idx++)
    {
        fd=&epx_thrd->events[event_idx];
        slot_idx=epx_thrd->events[event_idx].data.fd;
        slot=&epx->slots[slot_idx];
        if(epx->is_multi_thread) pthread_mutex_lock(&slot->lock);
        if(slot->slot)
        {
            if(epx_thrd->events[event_idx].events & EPOLLIN)
                _append_active_list(&head, &tail,
                        _x_fill(epx, epx_thrd, slot));
            if(epx_thrd->events[event_idx].events & EPOLLOUT)
                _x_flush(slot);
        }
        if(epx->is_multi_thread) pthread_mutex_unlock(&slot->lock);
        /*
           printf("events: %d\n",
           EPOLL_DATA.events[event_idx].events);
           printf("slot: %d\n", slot);
           */
    }
#elif defined(__FreeBSD__)
KQUEUE_AGAIN:
    timeout.tv_sec=curr_timestamp(NULL)+_POLL_TIMEOUT_MS / 1000;
    timeout.tv_nsec=(_POLL_TIMEOUT_MS % 1000) * 1000000;
    poll_num=kevent(epx->queue, NULL, 0, epx_thrd->events, 
            _FETCH_EVENT_NUM, &timeout);
    if(poll_num == SOCKET_ERROR)
    {
        if(get_last_error(NULL) == EINTR)
            goto KQUEUE_AGAIN;
        return(-2);
    }
    if(poll_num == 0)  return(0);

    /*
       printf("thrd %u\n", (unsigned int)pthread_self());
       printf("poll_num: %d\n", poll_num); 
       printf("active_num: %d\n", active_num);
       */
    for(event_idx=0; event_idx<poll_num; event_idx++)
    {
        slot_idx=epx_thrd->events[event_idx].udata;
        slot=&epx->slots[slot_idx];
        if(epx->is_multi_thread) pthread_mutex_lock(&slot->lock);
        if(epx_thrd->events[event_idx].events.filter == EVFILT_READ)
            _append_active_list(&head, &tail,
                    _x_fill(epx, epx_thrd, slot));
        else if(epx_thrd->events[event_idx].events.filter ==
                EVFILT_WRITE)
            _x_flush(slot);
        if(epx->is_multi_thread) pthread_mutex_unlock(&slot->lock);
        /*
           printf("events: %d\n",
           KQUEUE_DATA.events[event_idx].filter);
           printf("slot: %d\n", slot_idx);
           */
    }
#endif

    *active_list=head;
    return(0);
}/*}}}*/
        
int epx_write(EPX epx, EPX_SLOT epx_slot,
        char *msg, int *msg_size, unsigned short check)
{/*{{{*/
    struct _x_slot *slot;
    int msg_total_size, msg_left_size;
	int written_num;
    struct _write_node *write_node;
    int ret;

    if(!epx || !epx_slot || !msg || !msg_size)
        return(-1);
    msg_total_size=msg_left_size=*msg_size;
    if(msg_total_size == 0)  return(0);
    if(msg_total_size < 0 || msg_total_size > SOCK_MSG_SIZE)
        return(-1);

    if(epx->is_multi_thread) pthread_mutex_lock(&slot->lock);

    if(!epx_slot)
    {
        //printf("%d NULL thrd(%d) check(%d) slot_check(%d) msg_size(%d)\n", slot->slot_idx, thrd->thrd_idx, (int)check, (int)slot->check, *msg_size);
        ret=-2;
        goto ERR_OUT;
    }

    slot=&epx->slots[epx_slot->slot_idx];
    if(epx_slot->is_wrap)
    {
        wrap_size((unsigned char *)msg, &msg_total_size,
                (unsigned char *)msg);
        msg_left_size=msg_total_size;
    }

    if(check && check != slot->check)
    {
        //printf("CHECK slot_idx(%d) check(%d) slot_check(%d) msg_size(%d)\n", slot->slot_idx, (int)check, (int)slot->check, *msg_size);
        ret=-3;
        goto ERR_OUT;
    }

	// 写队列为空
    written_num=0;
	if(listp_is_empty(&epx_slot->write_list))
	{
AGAIN:
        written_num=send(epx_slot->sock, msg, msg_left_size, 0);
		if(written_num == SOCKET_ERROR)
		{
            int error=get_last_error(NULL);
		    if(error == EINTR)
                goto AGAIN;
			if(error != EAGAIN)
            {
                //printf("WRITE error(%d) slot_idx(%d) msg_size(%d)\n", error, slot->slot_idx, *msg_size);
                ret=-4;
                _x_slot_revoke(epx, slot);
                goto ERR_OUT;
            }
			written_num=0;
		}
		msg_left_size-=written_num;
		if(msg_left_size == 0)
            goto OUT_OK;
	}

	// 加入写队列
    ALLOC(write_node, struct _write_node, ERR_WRITE_NODE); 
    ALLOC_MULTI(write_node->msg, char, msg_left_size+1, ERR_WRITE_NODE);
    memcpy(write_node->msg, msg+written_num, msg_left_size);
    write_node->msg_size=msg_left_size;
    write_node->msg_written_size=0;
    listp_add_tail(&write_node->link, &epx_slot->write_list);

OUT_OK:
    if(epx->is_multi_thread) pthread_mutex_unlock(&slot->lock);
    return(0);

ERR_WRITE_NODE:
    FREE(write_node);
    ret=-5;

ERR_OUT:
    if(epx->is_multi_thread) pthread_mutex_unlock(&slot->lock);
    return(ret);
}/*}}}*/

#endif
